package com.xbai.spark.core

import java.sql.{Connection, PreparedStatement}

import org.apache.spark.rdd.{JdbcRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * MySQL数据库连接
  * 支持通过 Java JDBC 访问关系型数据库。需要通过 JdbcRDD 进行
  *   前提：添加 jdbc 连接依赖
  *
  * @author xbai
  * @Date 2021/1/4
  */
object Spark_Mysql {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("mysql")
    val sc = new SparkContext(conf)

    val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://47.102.136.26:3306/rdd"
    val userName = "root"
    val password = "123456"
    val sql = "select name, age from user where id >= ? and id <= ?"

    // 创建 jdbc RDD
    val jdbcRDD: JdbcRDD[Unit] = new JdbcRDD(sc, () => {
      // 获取数据库连接对象
      Class.forName(driver)
      java.sql.DriverManager.getConnection(url, userName, password)
    },
      sql,
      1,
      3,
      2,
      (res) => println(res.getString(1) + "," + res.getInt(2))
    )
    jdbcRDD.collect()

    val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("xbai", 23), ("xiaoming", 23)))
    dataRDD.foreachPartition(datas => {
      Class.forName(driver)
      val connection: Connection = java.sql.DriverManager.getConnection(url, userName, password)
      datas.foreach({
        case (userName, age) => {
          val sql = "insert into user(name, age) values(?,?)"
          val statement: PreparedStatement = connection.prepareStatement(sql)
          statement.setString(1, userName)
          statement.setInt(2, age)
          statement.executeUpdate()
          statement.close()
        }
      })
      connection.close()
    })

    sc.stop()
  }
}
